package com.techsun.datanow.sync

import com.microsoft.azure.cosmosdb.spark.config.Config
import com.microsoft.azure.cosmosdb.spark.schema._
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._


class Sync_TsrCustomerInfo extends TDataSync {
  override def sync(spark: SparkSession, cosmosDbEndpoint: String, cosmosDbMasterkey: String, kuduMaster: String, beginTime: Long, endTime: Long): Boolean = {

    try {
      Logger.getLogger("org").setLevel(Level.ERROR)
      Logger.getLogger("com").setLevel(Level.ERROR)
      val kuduContext = new KuduContext(kuduMaster, spark.sparkContext)

      val tableName = "sh_tsrcustomerinfo"

      val config = Config(Map(
        "Endpoint" -> cosmosDbEndpoint,
        "Masterkey" -> cosmosDbMasterkey,
        "Database" -> "VF_Vans_CN",
        "preferredRegions" -> "China East 2",
        "Collection" -> "TsrCustomerInfo",
        "query_custom" ->
          s"""
             |select
             |c.id,
             |c.source,
             |c.subSource subsource,
             |c.name,
             |c.gender,
             |c.birthday,
             |c.birthyear,
             |c.birthmonth,
             |c.birthdate,
             |c.birthDateChange birthdatechange,
             |c.phone,
             |c.mixphone,
             |c.isTncChecked  istncchecked,
             |c.shoesize,
             |c.interests,
             |c.brandTenure brandtenure,
             |c.provincecode,
             |c.province,
             |c.citycode,
             |c.city,
             |c.address,
             |c.email,
             |c.career,
             |c.education,
             |c.family,
             |c.isEmployee isemployee,
             |c.enrollState enrollstate,
             |c.isSendSms issendsms,
             |c.isSendWechat issendwechat,
             |c.isSendEmail issendemail,
             |c.isSendmm issendmm,
             |c.memo,
             |c.createTime createtime,
             |c.updateTime updatetime,
             |c.reservedField1 reservedfield1,
             |c.reservedField2 reservedfield2,
             |c.reservedField3 reservedfield3,
             |c.reservedField4 reservedfield4,
             |c.reservedField5 reservedfield5,
             |c._ts
             |from c where c._ts > ${beginTime}
             |""".stripMargin
      ))

      val schema = StructType(
        List(StructField("id", StringType, true),
          StructField("source", IntegerType, true),
          StructField("subsource", IntegerType, true),
          StructField("name", StringType, true),
          StructField("gender", IntegerType, true),
          StructField("birthday", LongType, true),
          StructField("birthyear", IntegerType, true),
          StructField("birthmonth", IntegerType, true),
          StructField("birthdate", IntegerType, true),
          StructField("birthdatechange", StringType, true),
          StructField("phone", StringType, true),
          StructField("mixphone", StringType, true),
          StructField("istncchecked", IntegerType, true),
          StructField("shoesize", StringType, true),
          StructField("interests", StringType, true),
          StructField("brandtenure", StringType, true),
          StructField("provincecode", IntegerType, true),
          StructField("province", StringType, true),
          StructField("citycode ", IntegerType, true),
          StructField("city", StringType, true),
          StructField("address", StringType, true),
          StructField("email", StringType, true),
          StructField("career", StringType, true),
          StructField("education", StringType, true),
          StructField("family", StringType, true),
          StructField("isemployee", IntegerType, true),
          StructField("enrollstate", IntegerType, true),
          StructField("issendsms", IntegerType, true),
          StructField("issendwechat", IntegerType, true),
          StructField("issendemail", IntegerType, true),
          StructField("issendmm", IntegerType, true),
          StructField("memo", StringType, true),
          StructField("createtime", TimestampType, true),
          StructField("updatetime", TimestampType, true),
          StructField("reservedfield1", StringType, true),
          StructField("reservedfield2", StringType, true),
          StructField("reservedfield3", StringType, true),
          StructField("reservedfield4", StringType, true),
          StructField("reservedfield5", StringType, true),
          StructField("_ts", IntegerType, true)
        ))

      val df = spark.sqlContext.read.schema(schema).cosmosDB(config)
      val count = df.count();
      printf("Sync_TsrCustomerInfo query count: %d\n", count)
      kuduContext.upsertRows(df, tableName)
      return true
    } catch {
      case ex: Exception => {
        println("Sync_TsrCustomerInfo exception: " + ex.getMessage)
        return false
      }
    }
  }
}
